package com.example.cloud.config;

import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.annotation.AnnotationAwareOrderComparator;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.Order;
import org.springframework.core.type.StandardMethodMetadata;

import com.example.cloud.config.interceptor.GRpcGlobalInterceptor;

import io.grpc.BindableService;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.ServerInterceptor;
import io.grpc.ServerInterceptors;
import io.grpc.ServerServiceDefinition;
import io.grpc.health.v1.HealthCheckResponse;
import io.grpc.services.HealthStatusManager;

/**
 * 管理GRPC Server
 * 
 * @author tongyufu
 *
 */
public class GRpcServerRunner implements ApplicationRunner, DisposableBean {

    Logger log = Logger.getLogger(getClass());
    @Autowired
    private HealthStatusManager healthStatusManager;
    @Autowired
    private AbstractApplicationContext applicationContext;
    private GRpcServerBuilderConfigurer configurer;
    private Server server;
    private final ServerBuilder<?> serverBuilder;

    public GRpcServerRunner(GRpcServerBuilderConfigurer configurer,
                            ServerBuilder<?> serverBuilder) {
        this.configurer = configurer;
        this.serverBuilder = serverBuilder;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("启动 GRPC Server ...");
        Collection<ServerInterceptor> globalInterceptors = getBeanNamesByTypeWithAnnotation(
            GRpcGlobalInterceptor.class, ServerInterceptor.class).map(
                name -> applicationContext.getBeanFactory().getBean(name, ServerInterceptor.class))
                .collect(Collectors.toList());

        serverBuilder.addService(healthStatusManager.getHealthService());

        getBeanNamesByTypeWithAnnotation(GRpcService.class, BindableService.class).forEach(name -> {
            BindableService service = applicationContext.getBean(name, BindableService.class);
            ServerServiceDefinition serviceDefinition = service.bindService();
            GRpcService gRpcServiceAnn = applicationContext.findAnnotationOnBean(name,
                GRpcService.class);
            serviceDefinition = this.bindInterceptors(serviceDefinition, gRpcServiceAnn,
                globalInterceptors);
            serverBuilder.addService(serviceDefinition);
            String serviceName = serviceDefinition.getServiceDescriptor().getName();
            healthStatusManager.setStatus(serviceName, HealthCheckResponse.ServingStatus.SERVING);
            log.info("注册GRPC服务：" + serviceName);
        });

        configurer.configure(serverBuilder);
        server = serverBuilder.build().start();
        log.info("GRPC服务已启动，监听端口：" + server.getPort());
        startDaemonAwaitThread();
    }

    @Override
    public void destroy() throws Exception {
        log.info("关闭GRPC Server...");
        server.getServices()
            .forEach(def -> healthStatusManager.clearStatus(def.getServiceDescriptor().getName()));
        Optional.ofNullable(server).ifPresent(Server::shutdown);
        log.info("GRPC Server已关闭");
    }

    private ServerServiceDefinition bindInterceptors(ServerServiceDefinition serviceDefinition,
                                                     GRpcService gRpcService,
                                                     Collection<ServerInterceptor> globalInterceptors) {

        Stream<? extends ServerInterceptor> privateInterceptors = Stream
            .of(gRpcService.interceptors()).map(interceptorClass -> {
                try {
                    return 0 < applicationContext.getBeanNamesForType(interceptorClass).length
                        ? applicationContext.getBean(interceptorClass)
                        : interceptorClass.newInstance();
                } catch (Exception e) {
                    throw new BeanCreationException("Failed to create interceptor instance.", e);
                }
            });

        List<ServerInterceptor> interceptors = Stream
            .concat(gRpcService.applyGlobalInterceptors() ? globalInterceptors.stream()
                : Stream.empty(), privateInterceptors)
            .distinct().sorted(serverInterceptorOrderComparator()).collect(Collectors.toList());
        return ServerInterceptors.intercept(serviceDefinition, interceptors);
    }

    private Comparator<Object> serverInterceptorOrderComparator() {
        Function<Object, Boolean> isOrderAnnotated = obj -> {
            Order ann = obj instanceof Method
                ? AnnotationUtils.findAnnotation((Method) obj, Order.class)
                : AnnotationUtils.findAnnotation(obj.getClass(), Order.class);
            return ann != null;
        };
        return AnnotationAwareOrderComparator.INSTANCE.thenComparing((o1, o2) -> {
            boolean p1 = isOrderAnnotated.apply(o1);
            boolean p2 = isOrderAnnotated.apply(o2);
            return p1 && !p2 ? -1 : p2 && !p1 ? 1 : 0;
        }).reversed();
    }

    private void startDaemonAwaitThread() {
        Thread awaitThread = new Thread(() -> {
            try {
                GRpcServerRunner.this.server.awaitTermination();
            } catch (InterruptedException e) {
                log.error("gRPC server stopped.", e);
            }
        });
        awaitThread.setDaemon(false);
        awaitThread.start();
    }

    private <T> Stream<String> getBeanNamesByTypeWithAnnotation(Class<? extends Annotation> annotationType,
                                                                Class<T> beanType) throws Exception {

        return Stream.of(applicationContext.getBeanNamesForType(beanType)).filter(name -> {
            final BeanDefinition beanDefinition = applicationContext.getBeanFactory()
                .getBeanDefinition(name);
            final Map<String, Object> beansWithAnnotation = applicationContext
                .getBeansWithAnnotation(annotationType);

            if (!beansWithAnnotation.isEmpty()) {
                return beansWithAnnotation.containsKey(name);
            } else if (beanDefinition.getSource() instanceof StandardMethodMetadata) {
                StandardMethodMetadata metadata = (StandardMethodMetadata) beanDefinition
                    .getSource();
                return metadata.isAnnotated(annotationType.getName());
            }

            return false;
        });
    }

}
